package org.goldenport.g3.drivers

import scala.collection.JavaConversions._
import scala.collection.mutable.ArrayBuffer
import org.goldenport.util.QSymbol
import org.goldenport.g3._
import org.goldenport.g3.adapters.{G3DataStoreDriver, G3DataStoreDriverAgent}
import org.goldenport.g3.messages._
import org.goldenport.g3.messages.datastore._
import org.goldenport.g3.messages.query.{IdPattern, RecordQuery, RecordQueryPattern}
import org.goldenport.g3.events._
import org.goldenport.g3.events.datastore._
import org.goldenport.wpath._
import com.google.appengine.api.datastore._
import com.google.appengine.api.datastore.Query.FilterOperator

/**
 * @since   Sep. 28, 2010
 * @version Jun. 22, 2011
 * @author  ASAMI, Tomoharu
 */
class AppEngineDataStoreDriver(val uri: String) extends G3DataStoreDriver {
  override protected def driver_Activity(ctx: G3AgentContext): PartialFunction[AnyRef, AnyRef] = {
    case cmd: G3Command => execute(cmd, ctx)
  }

  protected def new_Agent(cmd: G3Command, ctx: G3AgentContext): AppEngineDataStoreDriverAgent = {
    new AppEngineDataStoreDriverAgent(uri, ctx, recordClasses)
  }
}

class AppEngineDataStoreDriverAgent(
  uri: String, context: G3AgentContext,
  recordClasses: RecordClassSpace
) extends G3DataStoreDriverAgent(uri, context, recordClasses) {
  private val _datastore = DatastoreServiceFactory.getDatastoreService()
  private val _write_chunk_size = 1000

  def post(post: Post): G3Event = {
    insert_records(post, post.uri, post.asRecords)
  }

  def put(put: Put): G3Event = {
    update_records(put, put.uri, put.asRecords)
  }

  def mkcol(mkcol: Mkcol): G3Event = {
    error("AppEngineDataStoreDriver")
  }

  def create(msg: CreateKind) = {
    // do nothing?
    new DataStoreCreated(msg.uri, None)
  }

  def create(msg: CreateRecordClass) = {
    // do nothing?
    new DataStoreCreated(msg.uri, None)
  }

  def fetch(msg: FetchRecord) = {
    val entities = fetch_records(QSymbol(msg.uri), msg.ids)
    new DataStoreFetched(msg.uri, msg, entities, msg.ids.length == 1, None)
  }

  def query(msg: QueryRecord) = {
    val kind = QSymbol(msg.uri)
    val query = msg.query
    val patterns = query.patterns
    val result = patterns.collect {
      case id: IdPattern => id.ids
    } flatten match {
      case Nil => query_without_ids(kind, query, patterns)
      case ids: List[_] => query_with_ids(kind, query, ids)
    }
    new DataStoreQueried(msg.uri, msg, result, None)
  }

  def query_with_ids(kind: QSymbol, query: RecordQuery, ids: List[Any]): RecordSet = {
    fetch_records(kind, ids.map(_.toString))
    // XXX after filter and sorter
  }

  def query_without_ids(uri: QSymbol, query: RecordQuery,
                        patterns: List[RecordQueryPattern]): RecordSet = {
    val kind = get_kind(uri, "appengine")
    val q = new Query(kind)
    // XXX
    new AppEngineRecordSet(ds_query(q).iterator, context.context)
  }

  def query(msg: QueryRecordExample) = {
    val rs = query_example(msg.uri, msg.record)
    new DataStoreQueried(msg.uri, msg, rs, None)
  }

  private def query_example(msguri: String, example: Record): RecordSet = {
    val uri = QSymbol(msguri)
    val (kind, schema) = get_kind_schema(uri, "appengine")
    example.schema = schema // XXX side effect

    def query_with_id(id: AnyRef) = {
      fetch_records(uri, List(id.toString))
    // XXX after filter and sorter
    }

    def query_with_ids(ids: List[_]) = {
      fetch_records(uri, ids.map(_.toString))
      // XXX after filter and sorter
    }

    def query_without_ids() = {
      val q = new Query(kind)
      for ((k, v) <- example.fields) {
        v match {
//          case c: CValues => error("AppEngineDataStoreDriver")
//          case c: CDoubleRange => error("AppEngineDataStoreDriver")
//          case c: CLongRange => error("AppEngineDataStoreDriver")
          case c: Constraint => error("AppEngineDataStoreDriver")
          case _ => q.addFilter(k.name, FilterOperator.EQUAL, v)
        }
      }
      new AppEngineRecordSet(ds_query(q).iterator, context.context)
    }

    example.idOption match {
      case Some(id: Number) => query_with_id(id)
      case Some(id: String) => query_with_id(id)
      case Some(ids: List[_]) => query_with_ids(ids)
      case None => query_without_ids
    }
  }

  def query(msg: QueryRecordExpression) = {
    val rs = query_expression(msg.uri, msg.expression)
    new DataStoreQueried(msg.uri, msg, rs, None)
  }

  private def query_expression(msguri: String, expression: Expression) = {
    val uri = QSymbol(msguri)
    val expr = normalize_expression(uri, expression)
    expr match {
      case id: Id => fetch_records(uri, id.value.toString)
      case _ => error("AppEngineDataStoreDriver")
    }
  }

  def insert(msg: InsertRecord) = {
    val entity = to_entity(QSymbol(msg.uri))(msg.record)
    ds_put(entity)
    new DataStoreInserted(msg.uri, msg, RecordSet(msg.record), None)
  }

  def insert(msg: InsertRecordSet) = {
    insert_records(msg, msg.uri, msg.records)
  }

  private def insert_records(msg: G3Command, uri: String, records: RecordSet) = {
    // use var and assignment to release used elements of stream per loop
    var stream: Stream[Record] = records.records
    var done = false
    while (!done) {
      val (chunk, s) = get_chunk(stream)
      insert_chunk(QSymbol(uri), chunk)
      if (s.isEmpty) done = true
      stream = s
    }
    new DataStoreInserted(uri, msg, records, None)
  }

  private def get_chunk(stream: Stream[Record]): (Array[Record], Stream[Record]) = {
    var count = _write_chunk_size
    var s = stream
    val buf = new ArrayBuffer[Record]
    while (!s.isEmpty && count > 0) {
      buf += s.head
      s = s.tail
      count -= 1
    }
    (buf.toArray, s)
  }

  private def insert_chunk(kind: QSymbol, chunk: Array[Record]): java.util.List[Key] = {
    val entities: java.util.List[Entity] = chunk.map(to_entity(kind)).toList
    ds_put(entities)
  }

  def update(msg: UpdateRecord) = {
    val record = fetch_record(QSymbol(msg.uri), msg.record)
    record.setFrom(msg.record)
    val entity = to_entity(QSymbol(msg.uri))(record)
    ds_put(entity)
    new DataStoreUpdated(msg.uri, msg, RecordSet(msg.record), None)
  }

  def update(msg: UpdateRecordSet) = {
       update_records(msg, msg.uri, msg.records)
  }

  private def update_records(msg: G3Command, uri: String, records: RecordSet) = {
    // use var and assignment to release used elements of stream per loop
    var stream: Stream[Record] = records.records
    var done = false
    while (!done) {
      val (chunk, s) = get_chunk(stream)
      update_chunk(QSymbol(uri), chunk)
      if (s.isEmpty) done = true
      stream = s
    }
    new DataStoreUpdated(uri, msg, records, None)
  }

  private def update_chunk(kind: QSymbol, chunk: Array[Record]): java.util.List[Key] = {
    val records = fetch_records(kind, chunk.map(_.idString))
    for ((r, u) <- records zip chunk) {
      r.setFrom(u)
    }
    val entities: java.util.List[Entity] = chunk.map(to_entity(kind)).toList
    ds_put(entities)
  }

  def delete(msg: DeleteRecordId) = {
    delete_records(msg, msg.ids.map(_.toString))
  }

  def delete(msg: DeleteRecordExample) = {
    // TODO Optimize in case of id only example
    val rs = query_example(msg.uri, msg.record)
    delete_records(msg, rs.records.map(_.datastoreId.get.toString))
  }

  def delete(msg: DeleteRecordExpression) = {
    // TODO Optimize in case of id only example
    val rs = query_expression(msg.uri, msg.expression)
    delete_records(msg, rs.records.map(_.datastoreId.get.toString))
  }

  def drop(msg: DropKind) = {
    // XXX delete all datas?
    // call the generic kind remover via TaskQueue
    new DataStoreDropped(msg.uri)
  }

  //
  protected def fetch_Record(kind: QSymbol, id: String): Either[G3Event, Record] = {
    Right(fetch_record(kind, id))
  }

  private def fetch_record(uri: QSymbol, id: String): Record = {
    val (kind, idname) = get_kind_idname(uri, "appengine")
    val key = get_key(kind)(id)
    val entity = ds_get(key)
    to_record(entity)
  }

  protected def fetch_Records(kind: QSymbol, ids: List[Any]): Either[G3Event, RecordSet] = {
    Right(fetch_records(kind, ids))
  }

  private def fetch_records(uri: QSymbol, ids: Seq[Any]): RecordSet = {
    val (kind, idname) = get_kind_idname(uri, "appengine")
    val keys = ids.map(get_key(kind)_)
    val entities = ds_get(keys)
    val records = entities.map(e => to_record(e._2))
    RecordSet(records.toList) // XXX size problem
  }

  protected def query_Records(kind: QSymbol, query: WQuery): Either[G3Event, RecordSet] = {
    error("not implemented yet.")    
  }

  // get entity for update
  private def fetch_record(uri: QSymbol, record: Record): Record = {
    val (kind, idname) = get_kind_idname(uri, "appengine")
    find_key(kind, idname)(record) match {
      case Some(key) => to_record(ds_get(key))
      case None => throw new IllegalArgumentException("key not found")
    }
  }

  private def query_entities(kind: QSymbol, query: WQuery): RecordSet = {
    error("not implemented yet.")
  }

  override def delete(msg: Delete) = {
    val (kind, idname, id) = get_kind_idname_id(QSymbol(uri), "qppengine")
    val key = get_key(kind)(id)
    ds_delete(key)
    new DataStoreDeleted(msg.uri, msg, None)
  }

  private def delete_records(msg: G3Command, ids: Seq[String]) = {
    val (kind, idname) = get_kind_idname(QSymbol(msg.uri), "appengine")
    val keys = ids.map(get_key(kind)_)
    ds_delete(keys)
    new DataStoreDeleted(msg.uri, msg, None)
  }

  private def get_key(kind: String)(id: Any): Key = {
    try {
      val longid = id.asInstanceOf[AnyRef] match {
        case v: java.lang.Number => v.longValue
        case v => v.toString.toLong
      }
      kf_createKey(kind, longid)
    } catch {
      case e: NumberFormatException =>
        kf_createKey(kind, id.toString)
    }
  }

  private def find_key(kind: String, idname: Symbol)(record: Record): Option[Key] = {
    record.fields.find(_._1 == idname) collect {
      case (_, value) => get_key(kind)(value.toString)
    }
  }

  private def to_entity(uri: QSymbol)(record: Record): Entity = {
    val collect_fields = record.fields

    def collect_fields0 = {
      record.fields.collect {
        case (key, value) if key != 'id => (key, value)
      }
    }

    def setup_g3_properties(e: Entity) {
      e.setProperty("g3_version", "0.2") // XXX
      e.setProperty("g3_schema_uri", "schema:XXX")
    }

    def setup_atom_properties(e: Entity) {
      for (v <- record.atomId) {
        e.setProperty("atom_id", to_string(v))
      }
      for (v <- record.atomTitle) {
        e.setProperty("atom_title", to_string(v))
      }
      if (!record.atomCategories.isEmpty) {
        e.setProperty("atom_categories", to_string(record.atomCategories))
      }
      if (!record.atomAuthors.isEmpty) {
        e.setProperty("atom_authors", to_string(record.atomAuthors))
      }
      if (!record.atomContributers.isEmpty) {
        e.setProperty("atom_contributers", to_string(record.atomContributers))
      }
      for (v <- record.atomRights) {
        e.setProperty("atom_rights", to_string(v))
      }
      for (v <- record.atomSource) {
        e.setProperty("atom_source", to_string(v))
      }
      for (v <- record.atomSummary) {
        e.setProperty("atom_summary", to_string(v))
      }
      for (v <- record.atomContent) {
        e.setProperty("atom_content", to_string(v))
      }
      if (!record.atomLinks.isEmpty) {
        e.setProperty("atom_links", to_string(record.atomLinks))
      }
      for (v <- record.atomExtensionElements) {
        e.setProperty("atom_extension_elements", to_string(v))
      }

      val dateTime = VDateTime()
      record.atomPublished match {
        case Some(dt) => e.setProperty("atom_published", dt.toDate)
        case None => e.setProperty("atom_published", dateTime.toDate)
      }
      for (v <- record.atomUpdated) {
        e.setProperty("atom_updated", dateTime.toDate)
      }
    }

    val (kind, idname) = get_kind_idname(uri, "appengine")
    val entity = find_key(kind, idname)(record) match {
      case Some(key) => new Entity(key)
      case None => new Entity(kind)
    }
    for ((key, value) <- collect_fields) {
      entity.setProperty(key.name, to_entity_value(value, key, record.schema))
    }
    setup_g3_properties(entity)
    setup_atom_properties(entity)
    entity
  }

  private def to_record(entity: Entity): Record = {
    def is_app_property(key: String) = {
      key match {
        case "g3_version" => false
        case "g3_schema_uri" => false
        case "atom_id" => false
        case "atom_title" => false
        case "atom_updted" => false
        case "atom_published" => false
        case "atom_categories" => false
        case "atom_rights" => false
        case "atom_source" => false
        case "atom_summary" => false
        case "atom_content" => false
        case "atom_links" => false
        case "atom_extension_elements" => false
        case _ => true
      }
    }

    def app_properties = {
      entity.getProperties collect {
        case (key, value) if is_app_property(key) => {
          val atom = Symbol(key)
          (atom, to_record_value(value, atom, None))
        }
      } toList
    }

    def setup_system_properties(r: Record) {
      val key = entity.getKey
      key.getId match {
        case 0 => r.datastoreKey = Some(key.getName)
        case id => r.datastoreId = Some(id.asInstanceOf[AnyRef])
      }
      r.datastoreKey = Some(key)
    }

    def setup_g3_properties(r: Record) {
      for ((key, value) <- entity.getProperties) {
        key match {
          case "g3_version" => r.g3Version = Some(value.toString);
          case "g3_schema_uri" => r.g3SchemaUri = Some(value.toString)
          case _ => ;
        }
      }
    }

    def setup_atom_properties(r: Record) {
      for ((key, value) <- entity.getProperties) {
        key match {
          case "atom_id" => r.atomId = Some(to_string(value))
          case "atom_title" => r.atomTitle = Some(to_string(value))
          case "atom_updated" => r.atomUpdated = Some(to_datetime(value))
          case "atom_published" => r.atomPublished = Some(to_datetime(value))
          case "atom_categories" => r.atomCategories = to_string_list(value)
          case "atom_rights" => r.atomRights = Some(to_string(value))
          case "atom_source" => r.atomSource = Some(to_string(value))
          case "atom_summary" => r.atomSummary = Some(to_string(value))
          case "atom_content" => r.atomContent = Some(to_string(value))
          case "atom_links" => r.atomLinks = to_string_list(value)
          case "atom_extension_elements" => r.atomExtensionElements = to_xml_list(value)
          case _ => ;
        }
      }
    }

    val r = Record(app_properties)
    setup_system_properties(r)
    setup_g3_properties(r)
    r
  }

  private def to_string(value: Any) = value.toString

  private def to_string_list(value: Any): List[String] = {
    error("not implemented yet")
  }

  private def to_datetime(value: Any) = {
    VDateTime(value.asInstanceOf[java.util.Date])
  }

  private def to_xml_list(value: Any): List[scala.xml.Node] = {
    error("not implemented yet")
  }

  private def to_entity_value(v: AnyRef): AnyRef = {
    v match {
      case dt: VDateTime => dt.toDate
      case _ => v
    }
  }

  private def to_entity_value(v: AnyRef, key: Symbol, schema: Option[RecordSchema]): AnyRef = {
    schema match {
      case Some(sm) => {
        to_entity_value(v)
      }
      case None => to_entity_value(v)
    }
  }

  private def to_record_value(v: AnyRef, key: Symbol, schema: Option[RecordSchema]): AnyRef = {
    def without_schema(v: AnyRef) = {
      v match {
        case dt: java.util.Date => VDateTime(dt)
        case _ => v
      }
    }

    schema match {
      case Some(sm) => {
        without_schema(v)
      }
      case None => without_schema(v)
    }
  }

  //
  var retry_count = 10

  private def appengine_execute[A](f: => A): A = {
    val count = retry_count - 1
    for (i <- 0 until count) {
      try {
        return f
      } catch {
        case e: Exception => // XXX logging
      }
    }
    try {
      return f
    } catch {
      case e: Exception => throw e // XXX logging
    }
  }

  private def ds_get(key: Key): Entity = {
    appengine_execute {
      _datastore.get(key)
    }
  }

  private def ds_get(keys: java.lang.Iterable[Key]): java.util.Map[Key, Entity] = {
    appengine_execute {
      _datastore.get(keys)
    }
  }

  private def ds_put(entity: Entity): Key = {
    appengine_execute {
      _datastore.put(entity)
    }
  }

  private def ds_put(entities: java.lang.Iterable[Entity]): java.util.List[Key] = {
    appengine_execute {
      _datastore.put(entities)
    }
  }

  private def ds_delete(keys: java.lang.Iterable[Key]) {
    appengine_execute {
      _datastore.delete(keys)
    }
  }

  private def ds_delete(key: Key) {
    appengine_execute {
      _datastore.delete(key)
    }
  }

  private def ds_query(query: Query): QueryResultIterable[Entity] = {
    appengine_execute {
      _datastore.prepare(query).asQueryResultIterable
    }
  }

  private def kf_createKey(kind: String, id: Long) = {
    appengine_execute {
      KeyFactory.createKey(kind, id)
    }
  }

  private def kf_createKey(kind: String, name: String) = {
    appengine_execute {
      KeyFactory.createKey(kind, name)
    }
  }
}

class AppEngineRecordSet(val result: QueryResultIterator[Entity], val context: G3Context) extends StreamRecordSet {
  protected def load_Record(): Option[(Record, Stream[Record])] = {
    if (result.hasNext()) {
      val entity = result.next
      val record = new Record()
      for ((k, v) <- entity.getProperties) {
        record += Symbol(k) -> v // XXX data convert
      }
      Some((record,
           new AppEngineRecordSet(result, context)))
    } else {
      None
    }
  }
}
